package com.ylkj.stream.datahub;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.aliyun.datahub.client.model.*;
import com.hwl.datahub.out.IDatahubRecordResolver;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

/**
 * @Author: wengzx
 * @Date: 2025/4/2
 * @Desc: 优化后的 SinkResolver，根据 onlineStatus 构造不同的 content 输出
 */
@RequiredArgsConstructor
@Slf4j
public class SinkResolver implements IDatahubRecordResolver<String> {
    private RecordSchema recordSchema;

    // 初始化 RecordSchema，仅在 open 方法中设置一次
    @Override
    public void open() {
        recordSchema = new RecordSchema();
        recordSchema.addField(new Field("brokerip", FieldType.STRING));
        recordSchema.addField(new Field("clientid", FieldType.STRING));
        recordSchema.addField(new Field("data", FieldType.STRING));
        recordSchema.addField(new Field("publishdate", FieldType.STRING));
        recordSchema.addField(new Field("topic", FieldType.STRING));
        recordSchema.addField(new Field("deviceidentifier", FieldType.STRING));
        recordSchema.addField(new Field("eventtime", FieldType.STRING));
        recordSchema.addField(new Field("gatewayidentifier", FieldType.STRING));
        recordSchema.addField(new Field("serviceid", FieldType.STRING));
        recordSchema.addField(new Field("deviceid", FieldType.STRING));
        recordSchema.addField(new Field("gatewayid", FieldType.STRING));
        recordSchema.addField(new Field("headers", FieldType.STRING));
    }

    @Override
    public RecordEntry getRecordEntry(String vaule) {
        JSONObject inputObject = JSON.parseObject(vaule);
        RecordEntry recordEntry = new RecordEntry();
        TupleRecordData recordData = new TupleRecordData(recordSchema);
        // 提取必要字段
        String deviceCode = inputObject.getString("deviceCode");
        String onlineStatus = inputObject.getString("onlineStatus");
        String reportTime = inputObject.getString("reportTime");
        String topic = inputObject.getString("topic");
        if (StringUtils.isBlank(deviceCode) || StringUtils.isBlank(reportTime) || StringUtils.isBlank(topic)) {
            log.warn("必要字段缺失，跳过该记录。");
            return null;
        }

        // 构造 content 字段的 JSON 数据
        JSONObject contentJson = new JSONObject();
        contentJson.put("deviceId", deviceCode);
        contentJson.put("mid", 56); // 固定值
        JSONObject paramJson = new JSONObject();
        if ("online".equalsIgnoreCase(onlineStatus)) {
            paramJson.put("event", "online");
            paramJson.put("msg", "在线");
        } else if ("offline".equalsIgnoreCase(onlineStatus)) {
            paramJson.put("event", "offline");
            paramJson.put("msg", "下线");
        } else {
            log.warn("未知的 onlineStatus 值：{}，跳过该记录。", onlineStatus);
            return null;
        }
        paramJson.put("current", "");
        paramJson.put("limit", "");
        paramJson.put("container", "");
        paramJson.put("app", "");
        contentJson.put("param", paramJson);
        contentJson.put("timestamp", reportTime);
        contentJson.put("type", "EVENT_SYS_ALARM"); // 固定值
        recordData.setField("brokerip", null);
        recordData.setField("clientid", null);
        recordData.setField("data", contentJson.toJSONString()); // 设置 data 字段
        recordData.setField("publishdate", null);
        recordData.setField("topic", topic); // 设置 topic
        recordData.setField("deviceidentifier", null);
        recordData.setField("eventtime", null);
        recordData.setField("gatewayidentifier", null);
        recordData.setField("serviceid", null);
        recordData.setField("deviceid", null);
        recordData.setField("gatewayid", null);
        recordData.setField("headers", null);
        recordEntry.setRecordData(recordData);
        return recordEntry;
    }
}